package distributelock;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;

/**
 * zk分布式锁（避免羊群效应）
 */
public class DistributedLock implements Lock {
	
	private static final String zkServer = "127.0.0.1:2181";
	protected ZkClient zkClient = new ZkClient(zkServer);
	private static final String LOCK_PATH = "/LOCK";
	private CountDownLatch countDownLatch;
	private String beforePath; // 前一个节点
	private String currentPath; // 当前节点
	
	public DistributedLock() {
		if (!this.zkClient.exists(LOCK_PATH)) { // 判断有没有LOCK目录，没有则创建
            this.zkClient.createPersistent(LOCK_PATH);
        }
	}

	@Override
	public void lock() {
		if(tryLock()) { // 获取到锁
			System.out.println(Thread.currentThread().getName() + "获得分布式锁！");
		} else { // 获取锁失败
			waitForLock(); // 等待锁
			
			lock();  // 递归获取锁
		}
	}

	/**
	 * 获取锁
	 */
	@Override
	public boolean tryLock() {
		// 如果当前节点currentPath为空，则尝试获取锁
		if(this.currentPath==null || this.currentPath.length()==0) {
			// 创建一个临时节点
			this.currentPath = this.zkClient.createEphemeralSequential(LOCK_PATH + "/", "lock");
			System.out.println("currentPath:" + currentPath);
		}
		
		// 获取根节点下的所有临时节点并排序，临时节点名称为自增长的字符串
		List<String> children = this.zkClient.getChildren(LOCK_PATH);
		//由小到大排序所有子节点
        Collections.sort(children);
		
        // 判断刚才创建的子节点currentPath是否最小，如果是最小的，则获取锁，否则等待锁
        if(this.currentPath.equals(LOCK_PATH + "/" + children.get(0))) { // 当前节点是最小的
        	return true; // 获取到锁
        } else { // 当前节点不是最小的
        	// 找出比当前节点次小的节点(currentIndex-1)，作为beforePath
        	int currentIndex = Collections.binarySearch(children, currentPath.substring(6));
        	this.beforePath = LOCK_PATH + "/" + children.get(currentIndex-1);
        }
        
		return false;
	}

	/**
	 * 等待锁
	 */
	private void waitForLock() {
		System.out.println(Thread.currentThread().getName() + " 等待分布式锁......");
		
		// 创建节点监听器
		IZkDataListener listener = new IZkDataListener() {

			// 节点改变
			@Override
			public void handleDataChange(String dataPath, Object data) throws Exception {
				// TODO Auto-generated method stub
				
			}

			// 节点删除
			@Override
			public void handleDataDeleted(String dataPath) throws Exception {
//				System.out.println(Thread.currentThread().getName() + ":捕获到DataDelete事件！");
				if(countDownLatch!=null) {
					countDownLatch.countDown(); // 计数器-1
				}
			}
			
		};
		
		// 将节点监听器注册到前一个节点的watch事件上
		this.zkClient.subscribeDataChanges(beforePath, listener);
		
		// 判断beforePath节点是否存在，如果存在，创建一个计数器
		if(this.zkClient.exists(beforePath)) { // beforePath存在
			countDownLatch = new CountDownLatch(1);
			// 然后等待其他线程（只有它自己一个线程）
			try {
				countDownLatch.await();
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		
		// beforePath不存在，删除监听器
		this.zkClient.unsubscribeDataChanges(beforePath, listener);
	}

	/**
	 * 释放锁
	 */
	@Override
	public void unlock() {
		// 删除临时节点实现
//		if(this.zkClient.exists(currentPath)) {
			System.out.println(Thread.currentThread().getName() + "释放了锁---------");
			this.zkClient.delete(currentPath);
//		}
	}

	@Override
	public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
		return false;
	}

	@Override
	public void lockInterruptibly() throws InterruptedException {

	}

	@Override
	public Condition newCondition() {
		return null;
	}

}
